package com.hong.qrpc;

import com.alibaba.nacos.api.exception.NacosException;
import com.alibaba.nacos.api.naming.NamingFactory;
import com.alibaba.nacos.api.naming.NamingService;
import com.alibaba.nacos.api.naming.listener.EventListener;
import com.alibaba.nacos.api.naming.pojo.Instance;
import com.hong.config.constant.ClientConfig;
import com.hong.config.factory.ParserFactory;
import com.hong.config.parse.Parser;
import com.hong.qrpc.annotation.QRpcClient;
import com.hong.qrpc.codec.Decoder;
import com.hong.qrpc.codec.Encoder;
import com.hong.qrpc.config.ConfigManager;
import com.hong.qrpc.config.RegisterConfig;
import com.hong.qrpc.config.RpcClientConfig;
import com.hong.qrpc.factory.PropertiesStrategyFactory;
import com.hong.qrpc.listener.Listener;
import com.hong.qrpc.listener.ListenerManager;
import com.hong.qrpc.listener.RpcClientListener;
import com.hong.qrpc.listener.ServerListListener;
import com.hong.qrpc.listener.impl.Registrar;
import com.hong.qrpc.register.Register;
import com.hong.qrpc.selector.AbstractTransportSelector;
import com.hong.qrpc.utils.ReflectionUtils;
import com.hong.qrpc.utils.StringUtils;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;

import java.io.IOException;
import java.lang.reflect.Proxy;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @Author HQT
 * @Date 2022/7/10 19:40
 */
@Data
@Slf4j
public class RpcClient implements DisposableBean {
    private ConfigManager configManager = new ConfigManager();
    private Encoder encoder;
    private Decoder decoder;
    private ConfigurableListableBeanFactory beanFactory;
    private List<TransportSelector> directConnection = new LinkedList<>();
    /**
     * 每个服务名对应一个选择器，选择器中对所有的服务实例进行管理
     */
    private volatile ConcurrentHashMap<String, TransportSelector> selectors = new ConcurrentHashMap<>();
    private ListenerManager listenerManager = new ListenerManager();
    private Register register;

    public RpcClient(String profileUrl) throws IOException {
        init(profileUrl);
        afterRpcClientInit();
    }

    //不传url就全部走默认配置或者找默认路径
    public RpcClient() throws IOException {
        String defaultUrl = null;
        if (!StringUtils.isEmpty(defaultUrl = containDefaultProfile())) {
            init(defaultUrl);
        } else {
            init();
        }
        afterRpcClientInit();
    }

    /**
     * 在使用了springboot的项目中的服务端实例初始化
     *
     * @param remoteInvokes 需要添加相应的成员变量的代理对象
     * @param profile
     */
    public RpcClient(List<RemoteInvoke> remoteInvokes, Map<Object, Object> profile, ConfigurableListableBeanFactory beanFactory) throws IOException {
        this.beanFactory = beanFactory;
        loadCustomizationBean();
        configManager.setProfile(profile);
        initClientConfigProperties();
        Registrar.addRegistrar(this, remoteInvokes);
        init();
        afterRpcClientInit();
    }

    private String containDefaultProfile() {
        try {
            ResourceBundle.getBundle("qrpc.properties");
            return "qrpc.properties";
        } catch (Exception e) {
            return null;
        }
    }

    private void init(String profileUrl) throws IOException {
        ConfigManager configManager = new ConfigManager();
        String suffix = profileUrl.substring(profileUrl.lastIndexOf('.'));
        Parser parser = ParserFactory.getParser(suffix);
        configManager.setProfile(parser.parse(profileUrl));
        this.configManager = configManager;
        initClientConfigProperties();
        init();
    }

    public void initClientConfigProperties() {
        for (String propertiesName : ReflectionUtils.getAllConfigConstant(ClientConfig.class)) {
            PropertiesStrategyFactory.getPropertiesStrategy(propertiesName).
                    disposeProperties(this);
        }
    }

    private void init() {
        loadSpi();
        RpcClientConfig config = configManager.getRpcClientConfig();
        this.encoder = ReflectionUtils.newInstance(config.getEncoderClass());
        this.decoder = ReflectionUtils.newInstance(config.getDecoderClass());
    }

    private void afterRpcClientInit() {
        HashSet<Listener> listeners = listenerManager.getListenerSets().get(RpcClientListener.class);
        if (listeners != null) {
            for (Listener listener : listeners) {
                ((RpcClientListener) listener).afterRpcClientInit(this);
            }
        }
    }

    public Register getRegister() {
        return register;
    }

    public void setRegister(Register register) {
        this.register = register;
    }

    public ConfigManager getConfigManager() {
        return configManager;
    }

    public void setConfigManager(ConfigManager configManager) {
        this.configManager = configManager;
    }

    public TransportSelector newTransportSelector() {
        return ReflectionUtils.newInstance(configManager.getRpcClientConfig().getSelectorClass());
    }

    /**
     * 采用直连的方式进行调用，一次性使用，不予缓存
     *
     * @param clazz
     * @param serverAddr
     * @return
     */
    public <T> T getProxy(Class<T> clazz, String serverAddr) {
        String[] ipAndPort = StringUtils.splitByChars(serverAddr, ':');
        if (ipAndPort.length != 2) {
            throw new IllegalArgumentException("your serverAddr is illegal " + serverAddr);
        }
        TransportSelector selector = newTransportSelector();
        List<Peer> peers = Arrays.asList(new Peer(configManager.getRpcClientConfig().getProtocolType()
                , ipAndPort[0], Integer.parseInt(ipAndPort[1])));
        selector.init(peers, peers.size(), configManager.getRpcClientConfig().getTransportClass());
        RemoteInvoke remoteInvoke = new RemoteInvoke(clazz, encoder, decoder, selector);
        remoteInvoke.setSerializationType(configManager.getRpcClientConfig().getProtocolType());
        remoteInvoke.setTimeout(this);
        remoteInvoke.setRpcClient(this);
        remoteInvoke.setRetries(this);
        directConnection.add(selector);
        return (T) Proxy.newProxyInstance(
                clazz.getClassLoader(),
                new Class[]{clazz},
                remoteInvoke
        );
    }

    /**
     * 通过动态代理获取对象
     *
     * @param clazz 代理类
     * @return 代理类
     */
    public <T> T getProxy(Class<T> clazz) {
        QRpcClient anno = clazz.getAnnotation(QRpcClient.class);
        String serverName = anno.serviceName();
        if (StringUtils.isEmpty(serverName)) {
            throw new IllegalArgumentException(clazz.getName() + "of the annotation of serverName can't be \"\"");
        }
        //检查是否使用直连的方式，如果是的话就直连并缓存
        String addr = anno.addr();
        if (!StringUtils.isEmpty(addr)) {
            return getProxy(clazz, addr, serverName);
        }
        addSelectorIfAbsent(serverName);
        RemoteInvoke remoteInvoke = new RemoteInvoke(clazz, encoder, decoder, selectors.get(serverName));
        fullRemoteInvoker(remoteInvoke);
        return (T) Proxy.newProxyInstance(
                clazz.getClassLoader(),
                new Class[]{clazz},
                remoteInvoke
        );
    }

    /**
     * 该方法用于填充用于spring启动项目时为初始化完成的服务代理对象
     *
     * @param remoteInvoke
     */
    public void fullRemoteInvoker(RemoteInvoke remoteInvoke) {
        QRpcClient anno = (QRpcClient) remoteInvoke.getInterfaceClass().getAnnotation(QRpcClient.class);
        String serverName = anno.serviceName();
        addSelectorIfAbsent(serverName);
        remoteInvoke.setDecoder(decoder);
        remoteInvoke.setEncoder(encoder);
        remoteInvoke.setSelector(selectors.get(serverName));
        remoteInvoke.setSerializationType(configManager.getRpcClientConfig().getProtocolType());
        remoteInvoke.setTimeout(this);
        remoteInvoke.setRpcClient(this);
        remoteInvoke.setRetries(this);
    }

    private void loadRemoteAddrSelector(String addr, String serverName) {
        if (!selectors.containsKey(serverName)) {
            synchronized (selectors) {
                if (!selectors.containsKey(serverName)) {
                    TransportSelector selector = newTransportSelector();
                    String[] ipAndPort = StringUtils.splitByChars(addr, ':');
                    selector.init(Arrays.asList(new Peer(configManager.getRpcClientConfig().getProtocolType()
                                    , ipAndPort[0], Integer.parseInt(ipAndPort[1]))),
                            1, configManager.getRpcClientConfig().getTransportClass());
                    selectors.put(serverName, selector);
                }
            }
        }
    }

    /**
     * @param clazz      代理的接口类
     * @param addr       直连地址
     * @param serverName 服务名
     * @param <T>        代理对象
     * @return
     */
    private <T> T getProxy(Class<T> clazz, String addr, String serverName) {
        loadRemoteAddrSelector(addr, serverName);
        RemoteInvoke remoteInvoke = new RemoteInvoke(clazz, encoder, decoder, selectors.get(serverName));
        fullRemoteInvoker(remoteInvoke);
        return (T) Proxy.newProxyInstance(clazz.getClassLoader(),
                new Class[]{clazz},
                remoteInvoke);
    }

    public void addSelectorIfAbsent(String serverName) {
        //采用双锁检查懒加载的方式，只有需要的时候才会去拉对应发服务列表，拉完了会缓存在本地
        if (!selectors.containsKey(serverName)) {
            synchronized (selectors) {
                if (!selectors.containsKey(serverName)) {
                    RegisterConfig registerConfig = configManager.getRegisterConfig();
                    String registerAddrs = registerConfig.getRegisterAddrs();
                    if (StringUtils.isEmpty(registerAddrs)) {
                        return;
                    }
                    TransportSelector selector = newTransportSelector();
                    //初始化对应服务列表的监听器，每个服务名都会注册一个监听器用于监听服务变化
                    ServerListListener sl = ReflectionUtils.newInstance(configManager.getRegisterConfig().getServerListListenerClass());
                    sl.setRpcClientConfig(configManager.getRpcClientConfig());
                    sl.setSelector(selector);
                    //初始化服务器对象集合，准备进行选择器初始化
                    //去重一下
                    Set<Peer> peerDistinct = new HashSet<>();
                    for (String addr : StringUtils.splitByChars(registerAddrs, ';')) {
                        try {
                            NamingService service = NamingFactory.createNamingService(addr);
                            //过滤不健康实例
                            List<Instance> instances = service.selectInstances(serverName, true);
                            for (Instance instance : instances) {
                                Peer peer = new Peer(configManager.getRpcClientConfig().getProtocolType()
                                        , instance.getIp(), instance.getPort());
                                if (!peerDistinct.contains(peer)) {
                                    peerDistinct.add(peer);
                                }
                            }
                            //加载对应服务列表的监听器
                            NamingService naming = NamingFactory.createNamingService(addr);
                            naming.subscribe(serverName, (EventListener) sl);
                        } catch (NacosException e) {
                            e.printStackTrace();
                            log.warn("fail to find register center using addr:{}", addr);
                        }
                    }
                    List<Peer> peers = new ArrayList<>(peerDistinct);
                    selector.init(peers, peers.size(), configManager.getRpcClientConfig().getTransportClass());
                    selectors.put(serverName, selector);
                }
            }
        }
    }

    public void stop() {
        for (TransportSelector selector : selectors.values()) {
            selector.close();
        }
        selectors.clear();
        for (TransportSelector selector : directConnection) {
            selector.close();
        }
        directConnection.clear();
    }

    @Override
    public void destroy() throws Exception {
        stop();
    }

    public void loadCustomizationBean() {
        checkAndLoadSelector();
    }

    public void checkAndLoadSelector() {
        try {
            TransportSelector customizationSelector = beanFactory.getBean(AbstractTransportSelector.class);
            configManager.getRpcClientConfig().setSelectorClass(customizationSelector.getClass());
        } catch (Exception e) {
            //not find customization selector
        }
    }

    private <S> void loadSpi() {
        //加载负载均衡选择器
        for (AbstractTransportSelector selector : ServiceLoader.load(AbstractTransportSelector.class)) {
            configManager.getRpcClientConfig().setSelectorClass(selector.getClass());
        }
        //加载监听器
        for (RpcClientListener listener : ServiceLoader.load(RpcClientListener.class)) {
            Map<Class<? extends Listener>, HashSet<Listener>> sets = listenerManager.getListenerSets();
            HashSet<Listener> listeners = sets.get(RpcClientListener.class);
            if (listeners == null) {
                sets.put(RpcClientListener.class, listeners = new HashSet<>());
            }
            listeners.add(listener);
        }
    }
}
